package com.org.devg.elasticsearch;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

import java.io.IOException;
import java.lang.reflect.Array;
import java.lang.reflect.Field;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.org.devg.elasticsearch.wapper.JsonWapper;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig.Builder;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback;
import org.elasticsearch.client.RestClientBuilder.RequestConfigCallback;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.*;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.SearchModule;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;


public class EsSearch implements InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(EsSearch.class);

    private final static Lock esLock = new ReentrantLock();

    private static EsConfig esConfig;

    public EsSearch(EsConfig esConfig) {
        EsSearch.esConfig = esConfig;
    }

    public static <T> T getEsResultEntity(Map<String, Object> map, Class<T> sourceType) throws Exception {
        T t = sourceType.newInstance();
        Field[] fields = sourceType.getDeclaredFields();
        for (Field field : fields) {
            EsCloumn esCloumn = field.getAnnotation(EsCloumn.class);
            if (esCloumn != null) {
                field.setAccessible(true);
                String esName = esCloumn.name();
                boolean esArray = esCloumn.isArray();
                Object esValue = null;
                if (esArray) {
                    String arrayValue = JSON.toJSONString(map.get(esName));
                    Class<T> arrClazz = esCloumn.clazz();
                    if (null != arrayValue && !arrayValue.equals("")) {
                        JSONArray jsonArray = JSON.parseArray(arrayValue);
                        if (null == jsonArray || jsonArray.size() <= 0) {
                            continue;
                        }
                        int arraySize = jsonArray.size();
                        T[] arrDatas = (T[]) Array.newInstance(arrClazz, arraySize); //创建一个数组
                        for (int i = 0; i < arraySize; i++) {
                            T arrData = arrClazz.newInstance();
                            Field[] esFields = arrClazz.getDeclaredFields();
                            JSONObject jsonObject = jsonArray.getJSONObject(i);
                            for (Field arrfield : esFields) {
                                arrfield.setAccessible(true);
                                Object arrEsValue = getEsValue(arrfield, jsonObject);
                                Field arrEntityField = arrClazz.getDeclaredField(arrfield.getName());
                                arrEntityField.setAccessible(true);
                                arrEntityField.set(arrData, arrEsValue);
                            }
                            arrDatas[i] = arrData;
                        }
                        esValue = arrDatas;
                    }
                } else {
                    esValue = getEsValue(field, map);
                }
                Field entityField = t.getClass().getDeclaredField(field.getName());
                entityField.setAccessible(true);
                entityField.set(t, esValue);
                continue;
            }
        }
        return t;
    }

    public static Object getEsValue(Field field, Map<String, Object> map) {
        Object esValue = null;
        EsCloumn esCloumn = field.getAnnotation(EsCloumn.class);
        if (esCloumn != null) {
            String dateFormat = esCloumn.dateformat();
            String esName = esCloumn.name();
            String esTypeName = field.getType().getName();
            switch (esTypeName) {
                case "byte":
                    esValue = map.get(esName) == null ? null : Byte.valueOf(map.get(esName).toString());
                    break;
                case "java.lang.Byte":
                    esValue = map.get(esName) == null ? null : Byte.valueOf(map.get(esName).toString());
                    break;
                case "short":
                    esValue = map.get(esName) == null ? null : Short.valueOf(map.get(esName).toString());
                    break;
                case "java.lang.Short":
                    esValue = map.get(esName) == null ? null : Short.valueOf(map.get(esName).toString());
                    break;
                case "int":
                    esValue = map.get(esName) == null ? null : Integer.valueOf(map.get(esName).toString());
                    break;
                case "java.lang.Integer":
                    esValue = map.get(esName) == null ? null : Integer.valueOf(map.get(esName).toString());
                    break;
                case "long":
                    esValue = map.get(esName) == null ? null : Long.valueOf(map.get(esName).toString());
                    break;
                case "java.lang.Long":
                    esValue = map.get(esName) == null ? null : Long.valueOf(map.get(esName).toString());
                    break;
                case "char":
                    esValue = map.get(esName) == null ? null : map.get(esName).toString().charAt(0);
                    break;
                case "java.lang.Character":
                    esValue = map.get(esName) == null ? null : map.get(esName).toString().charAt(0);
                    break;
                case "float":
                    esValue = map.get(esName) == null ? null : Float.valueOf(map.get(esName).toString());
                    break;
                case "java.lang.Float":
                    esValue = map.get(esName) == null ? null : Float.valueOf(map.get(esName).toString());
                    break;
                case "double":
                    esValue = map.get(esName) == null ? null : Double.valueOf(map.get(esName).toString());
                    break;
                case "java.lang.Double":
                    esValue = map.get(esName) == null ? null : Double.valueOf(map.get(esName).toString());
                    break;
                case "java.math.BigDecimal":
                    esValue = map.get(esName) == null ? null : new BigDecimal(map.get(esName).toString());
                    break;
                case "java.util.Date":
                    esValue = map.get(esName) == null ? null : EsUtil.transferStringToDate(dateFormat, map.get(esName).toString());
                    break;
                default:
                    esValue = map.get(esName) == null ? null : String.valueOf(map.get(esName));
                    break;
            }
        }
        return esValue;
    }

    public static <T> List<T> getEsResultList(SearchHit[] searchHits, Class<T> sourceType) throws Exception {
        List<T> resultList = new ArrayList<>();
        if (null == searchHits || searchHits.length <= 0) {
            return resultList;
        }
        for (SearchHit hit : searchHits) {
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();
            T result = getEsResultEntity(sourceAsMap, sourceType);
            resultList.add(result);
        }
        return resultList;
    }

    public static <T> List<T> getEsResultList(SearchResponse searchResponse, Class<T> sourceType) throws Exception {
        SearchHits hits = searchResponse.getHits();
        SearchHit[] searchHits = hits.getHits();
        return getEsResultList(searchHits, sourceType);
    }

    public static <T> List<T> getEsResultList(SearchSourceBuilder searchSourceBuilder, Class<T> sourceType, String... indices) throws Exception {
        SearchHits hits = getSearchHits(searchSourceBuilder, indices);
        SearchHit[] searchHits = hits.getHits();
        return getEsResultList(searchHits, sourceType);
    }


    public static List<Map<String, Object>> getEsResultList(SearchSourceBuilder searchSourceBuilder, String... indices) throws Exception {
        List<Map<String, Object>> resultList = new ArrayList<>();
        SearchHits hits = getSearchHits(searchSourceBuilder, indices);
        SearchHit[] searchHits = hits.getHits();
        for (SearchHit hit : searchHits) {
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();
            resultList.add(sourceAsMap);
        }
        return resultList;
    }

    public static SearchRequest getSearchRequest(SearchSourceBuilder searchSourceBuilder, String... indices) {
        logger.info("indexName:{},elasticsearch-query:{}", indices, searchSourceBuilder.toString());
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.source(searchSourceBuilder);
        searchRequest.indices(indices);
        return searchRequest;
    }

    public static SearchResponse getSearchResponse(SearchSourceBuilder searchSourceBuilder, String... indices) throws IOException {
        searchSourceBuilder.trackTotalHits(true);
        SearchRequest searchRequest = getSearchRequest(searchSourceBuilder, indices);
        SearchResponse searchResponse = null;
        try {
            RestHighLevelClient client = esClient();
            searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        } catch (Exception ex) {
            logger.error("search-exception:", ex);
        }
        return searchResponse;
    }

    public static SearchHits getSearchHits(SearchSourceBuilder searchSourceBuilder, String... indices) throws IOException {
        return getSearchResponse(searchSourceBuilder, indices).getHits();
    }

    public static TotalHits getTotalHits(SearchSourceBuilder searchSourceBuilder, String... indices) throws IOException {
        return getSearchHits(searchSourceBuilder, indices).getTotalHits();
    }

    public static long getTotalValue(SearchSourceBuilder searchSourceBuilder, String... indices) throws IOException {
        return getTotalHits(searchSourceBuilder, indices).value;
    }

    public static Aggregations getAggregations(SearchSourceBuilder searchSourceBuilder, String... indices) throws IOException {
        return getSearchResponse(searchSourceBuilder, indices).getAggregations();
    }
    
    public static <T> EsResult<T> getEsResult(SearchSourceBuilder searchSourceBuilder, Class<T> sourceType, Integer pageNum, Integer pageSize,
                                              String... indices) throws Exception {
        searchSourceBuilder.from((pageNum - 1) * pageSize);
        searchSourceBuilder.size(pageSize);
        return getEsResult(searchSourceBuilder, sourceType, indices);
    }

    public static <T> EsResult<T> getEsResult(JsonWapper jsonWapper, Class<T> sourceType, String... indices) throws Exception {
        return getEsResult(jsonWapper.toString(), sourceType, indices);
    }

    public static <T> EsResult<T> getEsResult(String query, Class<T> sourceType, String... indices) throws Exception {
        SearchSourceBuilder searchSourceBuilder = getSearchSourceBuilder(query);
        return getEsResult(searchSourceBuilder, sourceType, indices);
    }

    public static <T> EsResult<T> getEsResult(SearchSourceBuilder searchSourceBuilder, Class<T> sourceType, String... indices) throws Exception {
        SearchResponse searchResponse = getSearchResponse(searchSourceBuilder, indices);
        SearchHits hits = searchResponse.getHits();
        List<T> esResultList = getEsResultList(hits.getHits(), sourceType);
        return EsResult.getEsResult(esResultList, searchResponse);
    }

    public static EsResult<Map<String, Object>> getEsResult(SearchSourceBuilder searchSourceBuilder, Integer pageNum, Integer pageSize,
                                                            String... indices) throws Exception {
        searchSourceBuilder.from((pageNum - 1) * pageSize);
        searchSourceBuilder.size(pageSize);
        return getEsResult(searchSourceBuilder, indices);
    }

    public static EsResult<Map<String, Object>> getEsResult(SearchSourceBuilder searchSourceBuilder, String... indices) throws Exception {
        SearchResponse searchResponse = getSearchResponse(searchSourceBuilder, indices);
        SearchHits hits = searchResponse.getHits();
        List<Map<String, Object>> resultList = new ArrayList<>();
        for (SearchHit hit : hits.getHits()) {
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();
            resultList.add(sourceAsMap);
        }
        return EsResult.getEsResult(resultList, searchResponse);
    }

    public static EsResult<Map<String, Object>> getEsResult(JsonWapper jsonWapper, String... indices) throws Exception {
        return getEsResult(jsonWapper.toString(), indices);
    }

    public static EsResult<Map<String, Object>> getEsResult(String query, String... indices) throws Exception {
        SearchSourceBuilder searchSourceBuilder = getSearchSourceBuilder(query);
        return getEsResult(searchSourceBuilder, indices);
    }

    public static SearchSourceBuilder getSearchSourceBuilder(String query) throws Exception {
        SearchModule searchModule = new SearchModule(Settings.EMPTY, false, Collections.emptyList());
        NamedXContentRegistry registry = new NamedXContentRegistry(searchModule.getNamedXContents());
        XContentParser xContentParser = XContentFactory.xContent(XContentType.JSON).createParser(registry, LoggingDeprecationHandler.INSTANCE, query);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.parseXContent(xContentParser);
        return searchSourceBuilder;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        initRestHighLevelClient();
    }

    public static RestHighLevelClient esClient() {
        RestHighLevelClient restHighLevelClient = EsMemory.restHighLevelClientMap.get(EsMemory.restHighLevelClient);
        if (null != restHighLevelClient) {
            return restHighLevelClient;
        }
        esLock.lock();
        try {
            initRestHighLevelClient();
            restHighLevelClient = EsMemory.restHighLevelClientMap.get(EsMemory.restHighLevelClient);
        } finally {
            esLock.unlock();
        }
        return restHighLevelClient;
    }

    public static void initRestHighLevelClient() {
        if (null == esConfig) {
            logger.error("init elasticsearch-config error....");
            return;
        }
        String hostName = esConfig.getHostname();
        if (null == hostName || hostName.trim().equals("")) {
            logger.error("init elasticsearch-host error....");
            return;
        }
        String port = esConfig.getPort();
        if (null == port || port.trim().equals("")) {
            logger.error("init elasticsearch-port error....");
            return;
        }
        String schemeName = esConfig.getSchemeName();
        if (null == schemeName || schemeName.trim().equals("")) {
            logger.error("init elasticsearch-scheme error....");
            return;
        }
        String[] hostArray = hostName.split(",");
        String[] portArray = port.split(",");
        String[] schemeArray = schemeName.split(",");
        if (hostArray.length != portArray.length || hostArray.length != schemeArray.length) {
            logger.error("init elasticsearch-httphost error....");
            return;
        }
        HttpHost[] httpHosts = new HttpHost[hostArray.length];
        for (int i = 0; i < hostArray.length; i++) {
            httpHosts[i] = new HttpHost(hostArray[i], Integer.valueOf(portArray[i]), schemeArray[i]);
        }
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(
                RestClient.builder(httpHosts)
                        .setRequestConfigCallback(new RequestConfigCallback() {
                            @Override
                            public Builder customizeRequestConfig(Builder requestConfigBuilder) {
                                requestConfigBuilder.setSocketTimeout(esConfig.getSocketTimeOut());
                                requestConfigBuilder.setConnectTimeout(esConfig.getConnectionTimeOut());
                                requestConfigBuilder.setConnectionRequestTimeout(esConfig.getConnectionRequestTimeOut());
                                return requestConfigBuilder;
                            }
                        }).setHttpClientConfigCallback(new HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
                        httpAsyncClientBuilder.setMaxConnTotal(esConfig.getMaxConnectTotal());
                        httpAsyncClientBuilder.setMaxConnPerRoute(esConfig.getMaxConnectPerRoute());
                        httpAsyncClientBuilder.setKeepAliveStrategy(CustomConnectionKeepAliveStrategy.INSTANCE);
                        if (!EsUtil.isNullOrEmpty(esConfig.getUserName()) && !EsUtil.isNullOrEmpty(esConfig.getUserPass())) {
                            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                            credentialsProvider.setCredentials(AuthScope.ANY,
                                    new UsernamePasswordCredentials(esConfig.getUserName(), esConfig.getUserPass()));
                            httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                        }
                        return httpAsyncClientBuilder;
                    }
                }));
        EsMemory.restHighLevelClientMap.put(EsMemory.restHighLevelClient, restHighLevelClient);
    }

}
